

-- ( 连接 DC data_cache mysql服务器专用文件 ) (  XServer[mysqlDcClient] --> DcSever[mysqlMgrDc] --> mysql )

local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log        = require("modules.commhm.log");
local ns_time    = require("modules.commhm.time");
local uu         = require("modules.commhm.utils");
local ns_network = require("modules.commhm.network");

local ns_centerServerProxy   = require("modules.commhm.centerServerProxy");

local table_insert = table.insert
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now


-------------------------------------------------------------------------------
--  __READ_ONLY__  只读   不用锁
-------------------------------------------------------------------------------


-- mysql MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


function _M.new(self, module_name_, config_server_, config_select_ )
	log.debug( log.fast() and "call mysqlDcClient:new");
	local ins_ = {
		config_cache  = {};
		module_name   = module_name_;

		config_server = config_server_;
		config_select = config_select_;

		carry_index = ngx.worker.id() * 10000 + 10000;
	};

	-- 异步上报config，失败不影响使用
	local config_str_ = table.tostring( { server=config_server_, select=config_select_ } )
	ns_centerServerProxy.WWW_configReport( module_name_ .. '_dc2', config_str_ )

    return setmetatable(ins_, mt);
end



--释放所有锁(远程)
function _M.final( self )
	--释放locks
	if  ngx.ctx.resty_dc_locks_mysql then
		for k, v in pairs( ngx.ctx.resty_dc_locks_mysql ) do
			if  v and v.select then
				if  v.cc > 0 then
					local ok, err_ = self:send_unlock_kh_to_dc( k, v.select, nil, v.lock_id )  --远程解锁
					if  ok then
						if  v.cc > 1 then
							log.day_list_simple( "dc_mysql", "lock_rw", "unlock_ok2", ngx.worker.id(),
									uu.get_act_cmd(), k, v.cc, debug.traceback() );
						end
					else
						log.day_list_simple( "dc_mysql", "lock_rw", "error_unlock2", ngx.worker.id(),
								uu.get_act_cmd(), k, v.cc, err_, debug.traceback() );
					end
				else
					--log.day_list_simple( "dc_mysql", "lock_rw", "unlocked2", ngx.worker.id(),
							--uu.get_act_cmd(), k, v.cc, debug.traceback() );
				end
			else
				log.day_list_simple( "dc_mysql", "lock_rw", "error_locks2", ngx.worker.id(),
						uu.get_act_cmd(), k, v.cc, debug.traceback() );
			end
		end
		ngx.ctx.resty_dc_locks_mysql = nil
	end

end


---解锁全部
function _M.unlock_all( self )
	self:final()
end


---解锁指定kh
function _M.unlock_key_hash( self, select_, key_, hash_ )
	local key_hash_ = key_ .. "_h" .. hash_
	self:send_unlock_kh_to_dc( key_hash_, select_ )
end


------------------------------------
---批量读取
_M.mread = function( self, select_, kv_list_, err_call_back )
	log.debug( log.fast() and "call mysqlDcClient:mread" );

	local kv_list_t_ = {}
	for i=1, #kv_list_ do
		kv_list_t_[ #kv_list_t_ + 1 ] =  kv_list_[i].key
		kv_list_t_[ #kv_list_t_ + 1 ] =  kv_list_[i].hash
	end

	if  #kv_list_t_ <= 0 then
		return {}
	end

	local kv_list_str_ = table.concat( kv_list_t_, '-' )

	local pp_ = {
		func    = 'mread',
		select  = select_,
		kv_list = kv_list_str_,
	}

	local ret_ = self:private_post_to_dc_server( pp_, err_call_back )
	local t_ = table.loadstring( ret_ ) or {}
	local fret_ = {}
	if  type(t_) == 'table' then
		for i=1, #t_ do
			if  type( t_[i] ) == 'table' and t_[i].v then
				-- { k1 hash v }
				if  t_[i].v == "not_found" then
					fret_[ #fret_ + 1 ] = ngx.null   --与原版本一致
				else
					fret_[ #fret_ + 1 ] = t_[i].v
				end
			end
		end
	end
	return fret_
end


----主动对数据进行加远程锁 (基本上不需要调用)
-- 一般read操作的时候会加锁， save或者final解锁
-- 特殊情况下需要直接加远程锁，目前只有ma服务在使用
function _M.add_read_lock( self, select_, key, hash, lock_id_ )
	local key_hash_ = key .. "_h" .. hash

	if  ngx.ctx.__READ_ONLY__ == 1 then
		---不用加锁
	else
		---是否已经被锁住了，如果已经锁了
		if  not ngx.ctx.resty_dc_locks_mysql then
			ngx.ctx.resty_dc_locks_mysql = {}
		end

		if  ngx.ctx.resty_dc_locks_mysql[ key_hash_ ] then
			if  ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc > 0 then
				--命中了cache，可以重复锁定
			else
				ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc = ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc + 1
			end
		else
			--新锁定成功
			ngx.ctx.resty_dc_locks_mysql [ key_hash_ ] = { cc=1, select=select_ }
			if  lock_id_ then
				ngx.ctx.resty_dc_locks_mysql [ key_hash_ ].lock_id = lock_id_
			end
			--log.day_list_simple( "ssdb", "lock_rw", "lock_ok",  ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ );
		end

		if  log.fast() then
			log.debug( "add resty_dc_locks_mysql1 : " .. key_hash_ )
			uu.var_dump( ngx.ctx.resty_dc_locks_mysql )
		end
	end
end


-- 加远程分布式锁 [key hash] or [kh] 二选一
_M.lock_kh = function(self, select, key_, hash_, hash_list_, err_call_back, et_, return_when_lock_ )
	local pp_ = {
		func    = 'lock_kh',
		select  = select,
		key     = key_,
		hash    = hash_,
		hash_list = hash_list_,

		lock_timeout = et_,                    --锁超时时间
		return_when_lock = return_when_lock_,  --发现已经锁定，是否不等待，直接返回locked
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


-- 解远程分布式锁
_M.unlock_kh = function(self, select, key_, hash_, hash_list_, lock_kh_id_, err_call_back )
	local pp_ = {
		func    = 'unlock_kh',
		select  = select,
		key     = key_,
		hash    = hash_,
		hash_list  = hash_list_,
		lock_kh_id = lock_kh_id_,    --通过lock_kh获得的锁id
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


----读取单个数据
--select=12345    【可选】分库分表id，不填则select值等于key值
--key=12345        key键值，可以是一个数字或者字符串
--hash=1           smallint（0到9999）   该name下的子分类id， 比如：1=背包 2=坐骑 3=皮肤
--err_call_back:  【可选】出错回调
--no_cache=y:     【可选】不读cache缓存，直接穿透读取mysql
--read_only=1:    【可选】表示只读请求，不加远程锁 外部设置 ngx.ctx.__READ_ONLY__ = 1
--lock_kh_id=10011  【可选】表示远程分布式锁
_M.read = function( self, select_, key, hash, err_call_back, no_cache_, read_only_, lock_kh_id_ )
	local pp_ = {
		func    = 'read',
		select  = select_,
		key     = key,
		hash    = hash,
		read_only = read_only_,
	}

	if  no_cache_ == 'y' then
		pp_.no_cache = 'y'
	end

	local key_hash_ = key .. "_h" .. hash

	if  ngx.ctx.__READ_ONLY__ == 1 then
		pp_.read_only = 1   --只读
	elseif  lock_kh_id_ then
		pp_.lock_kh_id = lock_kh_id_   --远程分布式锁
	else
		--是否已经被锁住了，如果已经锁了，转为一个只读请求
		if  not ngx.ctx.resty_dc_locks_mysql then
			ngx.ctx.resty_dc_locks_mysql = {}
		end
		if  ngx.ctx.resty_dc_locks_mysql[ key_hash_ ] and ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc > 0 then
			ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc = ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc + 1
			pp_.read_only = 1   --只读
			log.day_list_simple( "dc_mysql", "lock_rw", "relock_add_0", ngx.worker.id(), uu.get_act_cmd(), key_hash_,
					ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc, debug.traceback() );
		end
	end

	local ret_ = self:private_post_to_dc_server( pp_, err_call_back, 3000 )

	if  ret_ == '' then
		--返回空 (一般是获取锁失败) 重试
		if  ngx.ctx.__no_Content_Length__ then
			log.error( "private_post_to_dc_server retry " .. uu.get_act_cmd() ..  "\n" .. key_hash_ )
			ret_ = self:private_post_to_dc_server( pp_, err_call_back )
		else
			log.error( "private_post_to_dc_server has content-length, ignore retry " .. key_hash_ )
		end
	end

	if  pp_.read_only == 1 then
		--只读
	elseif lock_kh_id_ then
		--远程分布式锁
	elseif ret_ then
		--记录key_hash, 被锁定了
		if  not ngx.ctx.resty_dc_locks_mysql then
			ngx.ctx.resty_dc_locks_mysql = {}
		end

		if  ngx.ctx.resty_dc_locks_mysql[ key_hash_ ] then
			local lock_info_ = ngx.ctx.resty_dc_locks_mysql[ key_hash_ ]
			lock_info_.cc = lock_info_.cc + 1

			if  lock_info_.cc > 1 then
				--重复锁定
				log.day_list_simple( "dc_mysql", "lock_rw", "relock_add_1", ngx.worker.id(),
						uu.get_act_cmd(), key_hash_, lock_info_.cc, debug.traceback() );
			end

			if  ngx.ctx.__lock_id__ then
				if  lock_info_.lock_id == ngx.ctx.__lock_id__ then
					log.day_list_simple( "dc_mysql", "lock_rw", "relock_lock_id_match_ok", ngx.worker.id(),
							uu.get_act_cmd(), key_hash_, ngx.ctx.__lock_id__ )
				else
					log.day_list_simple( "dc_mysql", "lock_rw", "relock_lock_id_unmatch",  ngx.worker.id(),
							uu.get_act_cmd(), key_hash_,  lock_info_.lock_id, ngx.ctx.__lock_id__ )
					lock_info_.lock_id = ngx.ctx.__lock_id__
				end
			end

		else
			--新锁成功
			ngx.ctx.resty_dc_locks_mysql [ key_hash_ ] = { cc=1, select=select_ }
			if  ngx.ctx.__lock_id__ then
				ngx.ctx.resty_dc_locks_mysql [ key_hash_ ].lock_id = ngx.ctx.__lock_id__
			end
			--log.day_list_simple( "dc_mysql", "lock_rw", "lock_ok",  ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ );
		end

	end

	if  ret_ == "not_found" then
		log.debug( "not_found set to ngx.null" )
		ret_ = ngx.null          --与原版本一致
	end

	return ret_
end



----写入单个数据，并解开远程锁
--select:         用来做十分表或者百分表的分表id，一般是uin
--key:            一般是uin
--hash:           同一个uin下的分类，参考 const_vars.lua 里的FOO.SSDB.KEYS
--err_call_back:  【可选】出错回调
--delay_       :  【可选】1=不立即写mysql   其他(默认值)=立即同时写mysql
--read_only_   :  【可选】1=不解开远程锁     其他(默认值)=如果有远程锁，立即解锁
--                  （可作为主动参数传入=1，暂时不用解锁，等待final解锁，适用于一次读取，多次写入的情况）
--ver_         :  【可选】数据版本锁 有且大于0的时候，需要比数据库版本值大才能写入成功
_M.save = function( self, select_, key, hash, value, err_call_back, delay_, read_only_, ver_, lock_kh_id_ )
	local pp_ = {
		func    = 'save',
		select  = select_,
		key     = key,
		hash    = hash,
		value   = value,
		read_only = read_only_,  --是否不解远程锁
		ver     = ver_,           --数据版本锁
	}

	if  delay_ == 1 then
		pp_.mysql_delay = 1  --是否不立即写mysql
	end

	if  read_only_ == 1 then
		-- 只读强制写，不判断
	elseif  ngx.ctx.__READ_ONLY__ == 1 then
		--只读接口有写操作
		log.day_list_simple( "dc_mysql", "lock_rw", "error_save_where_read_only", ngx.worker.id(),
				uu.get_act_cmd(), key .. "_h" .. hash, debug.traceback() );
	end

	local key_hash_ = key .. "_h" .. hash
	if  read_only_ == 1 then
		---不用解锁
	elseif lock_kh_id_ then
		pp_.lock_kh_id = lock_kh_id_
	else
		---获得解锁id
		if  ngx.ctx.resty_dc_locks_mysql and
				ngx.ctx.resty_dc_locks_mysql[ key_hash_ ] and
				ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].lock_id then
			pp_.lock_id = ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].lock_id
		end
	end

	--写操作 判断一下锁
	local ret_ = self:private_post_to_dc_server( pp_, err_call_back, 3000 )
	if  ret_ then
		--试图解锁
		if  read_only_ == 1 then
			--不用解锁 ( 或者作为主动参数传入，暂时不用解锁，等待final解锁 )
		elseif lock_kh_id_ then
			--不用解锁，使用远程分布式锁
		elseif  ngx.ctx.resty_dc_locks_mysql and ngx.ctx.resty_dc_locks_mysql[ key_hash_ ] then
			if  ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc > 0 then
				ngx.ctx.resty_dc_locks_mysql[ key_hash_ ].cc = 0   --解锁成功
				--log.day_list_simple( "dc_mysql", "lock_rw", "unlock_ok",
						--ngx.worker.id(), uu.get_act_cmd(), 0, key_hash_ );
			else
				--重复解锁 使用缓存连写两次
				--log.day_list_simple( "dc_mysql", "lock_rw", "warn_re_unlock",  ngx.worker.id(),
						--uu.get_act_cmd(), 0,     key_hash_, debug.traceback() );
			end
		else
			--找不到锁
			log.day_list_simple( "dc_mysql", "lock_rw", "error_not_find_lock", ngx.worker.id(),
					    uu.get_act_cmd(), 'nil', key_hash_, debug.traceback() );
		end
	end
	return ret_
end


--- 删除一个key下的一个hash
_M.del = function( self, select, key, hash, err_call_back )
	local pp_ = {
		func    = 'del',
		select  = select,
		key     = key,
		hash    = hash,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--- 一个key下有多少个hash值
_M.hlist = function( self, select, start_, end_, size_, err_call_back)
	local pp_ = {
		func    = 'hlist',
		select  = select,
		start   = start_,
		h_end   = end_,
		size    = size_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--获得所有值
_M.hgetall = function( self, select, key_, err_call_back)
	local pp_ = {
		func    = 'hget_all',
		select  = select,
		key     = key_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end



--- 一个key下有多少个hash值
_M.hlen = function( self, select, key, err_call_back)
	local pp_ = {
		func    = 'hlen',
		select  = select,
		key     = key,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


---获得一个key下的所有值
_M.hget_all = function( self, select, key, err_call_back )
	local pp_ = {
		func    = 'hget_all',
		select  = select,
		key     = key,
	}
	return self:private_post_to_dc_server( pp_, err_call_back )
end


--[[---设置单个kv
_M.set_kv = function( self, select, key, value, et, err_call_back )
	local pp_ = {
		func    = 'set_kv',
		select  = select,
		key     = key,
		value   = value,
		et      = et,
	}

	return self:private_post_to_dc_server( pp_, err_call_back )
end


---获得单个kv
_M.get_kv = function( self, select, key, err_call_back )
	local pp_ = {
		func    = 'get_kv',
		select  = select,
		key     = key,
	}

	local ret_ = self:private_post_to_dc_server( pp_, err_call_back )
	if  ret_ == "not_found" then
		ret_ = ngx.null          --与原版本一致
	end
	return ret_
end]]


---发送远程解锁请求
_M.send_unlock_kh_to_dc = function( self, key_hash_, select_, err_call_back, lock_id_ )
	local pp_ = {
		func    = 'unlock_kh',
		select  = select_,
		kh      = key_hash_,
		lock_id = lock_id_,
	}
	return self:private_post_to_dc_server( pp_, err_call_back, 3000 )
end


---远程清理缓存，清理后会重新加载数据
_M.clean_shared_cache_ssdb = function(self, select_, key_, hash_ )
	local pp_ = {
		func    = 'clean_cache',
		select  = select_,
		key     = key_,
		hash    = hash_,
	}
	return self:private_post_to_dc_server( pp_, 3000 )
end


---发送请求到服务器 POST
_M.private_post_to_dc_server = function( self, pp_, err_call_back, p_time_exp_ )

	ngx.ctx.err_call_back = err_call_back;
	local url_ = self:private_get_url_by_select( pp_.select )

	self.carry_index = self.carry_index + 1
	local Carry_ = '' .. self.carry_index

	--http://120.24.64.132:8086/dc2/?auth=Xcn67K&name=team_mysql&cmd=read&carry=111&select=1&key=12345&hash=1
	url_ = url_ .. "&cmd=comm&carry=" .. Carry_ .. "&select=" .. pp_.select
	if  ngx.ctx.m_params.uin then
		url_ = url_ .. "&uin=" .. ngx.ctx.m_params.uin
	end

	local post_data_ = ngx.encode_args(pp_);
	local ret_, body_, resp_ = ns_network.postHttpPage( url_, post_data_, p_time_exp_ )

	if  log.fast() then
		local url_pos_ = string.find( url_, '?' )
		if  url_pos_ and url_pos_>0 then
			log.debug( "call private_post_to_dc_server: " .. string.sub( url_, 1, url_pos_-1 ) )
		else
			log.debug( "call private_post_to_dc_server: " .. url_ )
		end
		uu.var_dump( pp_ )
		log.debug( post_data_ )
		log.debug( "mysql post ret=" .. ret_ .. ", body=" .. (body_ or 'nil') )
	end

	if  ret_ == 200 then
		--验证Carry
		if  resp_.headers.Carry == Carry_ then
			---lock id
			if  pp_.func == 'read' and resp_.headers["LockId"] then
				ngx.ctx.__lock_id__ =  resp_.headers["LockId"]
				--log.day_list_simple( "dc_mysql", "dc_client", "get_read_lock_id",
					--ngx.worker.id(), ngx.ctx.__lock_id__ )
			end

			if  resp_.headers["Content-Length"] then
				if  resp_.headers["Content-Length"] ==  ('' .. #body_) then
					return body_   --ok
				else
					log.day_list_simple( "dc_mysql", "dc_client", "DataLen_check_error",
							resp_.headers["Content-Length"], #body_ )
				end
			else
				ngx.ctx.__no_Content_Length__ = 1
				return body_       --ok 没有Content-Length
			end
		else
			log.day_list_simple( "dc_mysql", "dc_client", "Carry_check_error", resp_.headers.Carry, #Carry_ )
		end
	else
		log.day_list_simple( "dc_mysql", "dc_client", "http_error", ret_, url_ )
	end

	self:error_exit( 'error select = ' .. (pp_.select or 'nil') );
end



---------------------------------------
---根据select_获得连接的url
_M.private_get_url_by_select = function( self, select_ )
	if  self.config_cache[ select_ ]  then
		--先查config cache
		log.debug( log.fast() and "find config_cache for " .. select_ );
		return  self.config_cache[ select_ ];
	else
		if  self.config_server and self.config_select then
			if  self.config_select[select_] and
				self.config_server[ self.config_select[select_] ]     and
				self.config_server[ self.config_select[select_] ].url then
				local ret_ = self.config_server[ self.config_select[select_] ].url;
				if  ret_ then
					---log.debug( log.fast() and "set config_cache for " .. select_ );
					self.config_cache[ select_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:error_exit( 'error select_ = ' .. (select_ or 'nil') );

end



---------------------------------------
-- 出错处理
_M.error_exit = function(self, txt, trace_info_ )
	if  ngx.ctx.err_call_back then
		--log.error( "has set error_exit. " .. (txt or 'nil') );
		ngx.ctx.err_call_back(txt, trace_info_);
	else
		log.error( "not set error_exit. " .. (txt or 'nil'), trace_info_ );
	end
end



---------------------------------------------校验函数----------------------------
--写校验[_k_]
_M.private_check_set_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "dc_mysql", "error_key_check", "set", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "dc_mysql", "error_hash_check", "set", select_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end


--读校验[_k_]
_M.private_check_get_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "dc_mysql", "error_key_check", "get", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "dc_mysql", "error_hash_check", "get", select_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end



return _M;
